/*
 * Copyright (c) 2018-2024 北京华恒盛世科技有限公司
 * QUICKPOOL is licensed under Mulan PSL v2.
 * You can use this software according to the terms and conditions of the Mulan PSL v2.
 * You may obtain a copy of Mulan PSL v2 at:
 *     http://license.coscl.org.cn/MulanPSL2
 * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
 * EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
 * MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
 * See the Mulan PSL v2 for more details.
*/

#include <unistd.h>
#include <stdio.h>
#include <time.h>
#include <pthread.h>
#include <ctype.h>
#include <netdb.h>
#include <fcntl.h>
#include <arpa/inet.h>
#include <errno.h>
#include <sys/epoll.h>
#include <sys/poll.h>
#include <rpc/xdr.h>

#include "common.h"
#include "hash.h"
#include "channel.h"

#define BIND_RETRY_TIMES 100

#define MAX_CONN     1024
#define LISTENQ      1024

static int maxchannels;
static struct channel *channels;
static HASH_TABLE_T *chanhash;

static pthread_mutex_t mutexChannels = PTHREAD_MUTEX_INITIALIZER;
static pthread_mutex_t mutexChanHash = PTHREAD_MUTEX_INITIALIZER;
static pthread_mutex_t mutexMasterName = PTHREAD_MUTEX_INITIALIZER;

static char *masterName;

static int epfd;
static int listenfd = -1;

static struct epoll_event *events;

static void free_data(void *data);

static struct channel* get_free_channel(int *index);
static struct channel* get_channel_at(int index);
static int add_chanidx(struct channel *channel, int index);
static int find_chanidx(int fd);

/**
 * @brief        To initialize the socket channels
 *
 * @retval       -1      #1: Failed
 * @retval        0      #2: Success
 *
 * @note         A hash table is created for quick search for the socket fd.
 *               errno is set when allocate memory failed.
 */
int
gf_chan_init(void)
{
    static char first = 1;

    if (first == 0) {
        return 0;
    }

    first = 0;
    
    maxchannels = sysconf(_SC_OPEN_MAX);
    if (maxchannels > 4096) {
        maxchannels = 4096;
    }

    channels = (struct channel *)calloc(maxchannels, sizeof(struct channel));
    if (channels == NULL) {
        log_message(ERR, NOCLASS, "%s: calloc(%d) failed: %m.",
                    __func__, maxchannels*sizeof(struct channel));
        return -1;
    }

    int i;
    for (i=0; i<maxchannels; i++) {
        channels[i].fd = -1;
        channels[i].state = CHANNEL_FREE;
        channels[i].senddatalist = gf_list_make("Channel sending data list");
        channels[i].recvdatalist = gf_list_make("Channel recerving data list");
    }

    chanhash = gf_hash_make(maxchannels);
    if (chanhash == NULL) {
        log_message(ERR, NOCLASS, "%s: make hash table failed.", __func__);
        return (-1);
    }

    epfd = epoll_create(maxchannels);
    events = (struct epoll_event *)calloc(maxchannels, sizeof(struct epoll_event));
    if (NULL == events) {
        log_message(ERR, NOCLASS, "%s: calloc(%d) failed: %m.",
                    __func__, maxchannels*sizeof(struct epoll_event));
        return -1;
    }

    return 0;
} /* gf_chan_init */


/**
 * @brief        To initialize the server listen socket
 *
 * @retval       -1      #1: Failed
 * @retval        0      #2: Success
 *
 * @note         A hash table is created for quick search for the socket fd.
 */
int 
gf_init_server(struct sockaddr_in *serveraddr)
{
    if (gf_chan_init() < 0) {
        return (-1);
    }

    listenfd = socket(AF_INET, SOCK_STREAM, 0);

    gf_set_nonblocking(listenfd);

    int reuse1 = 1;
    if (setsockopt(listenfd, SOL_SOCKET, SO_REUSEADDR, (char *)&reuse1, sizeof(reuse1))==-1) {
        log_message(ERR, NOCLASS, "%s: setsockopt failed when initialize server: %m.", __func__);
        return (-1);
    }
    reuse1 = 1;

    struct epoll_event ev;
    memset(&ev, 0, sizeof(struct epoll_event));            
    ev.data.fd = listenfd;
    ev.events = EPOLLIN|EPOLLET;
    epoll_ctl(epfd, EPOLL_CTL_ADD, listenfd, &ev);

    unsigned int len;
    len = sizeof(struct sockaddr);
    if (gf_sock_bind(listenfd, (struct sockaddr *)serveraddr, len) < 0) {
        log_message(ERR, NOCLASS, "%s: bind port[%d] failed due to %m.",
                    __func__, ntohs(serveraddr->sin_port));
        return (-1);
    }

    if (listen(listenfd, LISTENQ) < 0) {
        log_message(ERR, NOCLASS, "%s: listen on port[%d] failed due to %m.",
                    __func__, ntohs(serveraddr->sin_port));
        return (-1);
    }

    return 0;
} /* gf_init_server */

int
gf_get_sockport(void)
{
    struct sockaddr_in myAddr;
    int ret;
    unsigned int len;
    len = sizeof(struct sockaddr);
    ret = getsockname(listenfd, (struct sockaddr *)&myAddr, &len);
    if (0 < ret) {
        log_message(ERR, NOCLASS, "%s: getsockname(%d) failed due to %m.",
                    __func__, listenfd);
        return -1;
    }

    return (myAddr.sin_port);
}

int gf_get_epfd(void)
{
    return (epfd);
}

/**
 * @brief        Find a free channel and add it to the hash table
 *
 * @param[in]    fd   #1: The socket connection fd.
 *
 * @retval       -1      #1: Failed
 * @retval        0      #2: Success
 *
 * @note         The channel state is changed to CHANNEL_CONN.
 *               errno is set if there is no free channel.
 */
int
gf_new_conn(int fd)
{
    int index;

    index = find_chanidx(fd);
    if (index >= 0) {
        return (0);
    }

    struct channel *conn = get_free_channel(&index);
    if (conn == NULL) {
        return (-1);
    }

    pthread_mutex_lock(&mutexChannels);
    conn->fd = fd;
    conn->state = CHANNEL_CONN;
    add_chanidx(conn, index);
    pthread_mutex_unlock(&mutexChannels);

    return (0);
} /* gf_new_conn */

/**
 * @brief        close the socket fd and set the channel to free.
 *
 * @param[in]    fd   #1: The socket connection fd.
 *
 * @retval       none
 *
 * @note         The channel state is changed to CHANNEL_FREE.
 *               Freeup the old data in the channel and calloc memory
 *               for the upcoming data.
 *               Remove the channel from the hash table.
 */
void
gf_chan_close(int fd)
{
    if (fd < 0) {
        return;
    }

    int index = find_chanidx(fd);
    if (index < 0) {
        if (fd >= 0) {
            close(fd);
        }
        return;
    }

    struct channel *chan = get_channel_at(index);
    if (chan == NULL) {
        if (fd >= 0) {
            close(fd);
        }
        return;
    }

    pthread_mutex_lock(&mutexChanHash);
    char fdKey[12];
    snprintf(fdKey, 12, "%d", chan->fd);
    int *removeIdx = (int *)gf_hash_remove(chanhash, fdKey);
    FREEUP(removeIdx);
    pthread_mutex_unlock(&mutexChanHash);

    pthread_mutex_lock(&mutexChannels);
    chan->state = CHANNEL_FREE;
    gf_list_free(chan->senddatalist, free_data);
    chan->senddatalist = gf_list_make("Channal sending data list");

    gf_list_free(chan->recvdatalist, free_data);
    chan->recvdatalist = gf_list_make("Channal receiving data list");

    if (chan->fd >= 0) {
        close(chan->fd);
    }
    pthread_mutex_unlock(&mutexChannels);

    return;
} /* gf_chan_close */

void
gf_chan_close_accept(void)
{
    gf_chan_close(listenfd);
    gf_chan_close(epfd);

    return;
}

void
gf_child_chan_close(int except)
{
    int i;

    gf_chan_close(listenfd);

    for (i = 0; i < maxchannels; i++) {
        if (channels[i].state == CHANNEL_FREE) {
            continue;
        }

        if (channels[i].fd == except || channels[i].fd < 0) {
            continue;
        }

        pthread_mutex_lock(&mutexChanHash);
        char fdKey[12];
        snprintf(fdKey, 12, "%d", channels[i].fd);
        int *removeIdx = (int *)gf_hash_remove(chanhash, fdKey);
        FREEUP(removeIdx);
        pthread_mutex_unlock(&mutexChanHash);

        pthread_mutex_lock(&mutexChannels);
        channels[i].state = CHANNEL_FREE;
        gf_list_free(channels[i].senddatalist, free_data);
        channels[i].senddatalist = gf_list_make("Channal sending data list");

        gf_list_free(channels[i].recvdatalist, free_data);
        channels[i].recvdatalist = gf_list_make("Channal receiving data list");

        close(channels[i].fd);
        pthread_mutex_unlock(&mutexChannels);
    }

    return;
}

/**
 * @brief        Connet to remote server.
 *
 * @param[in]    host    #1: Name of the host to be connected.
 * @param[in]    port    #2: The destinition port.
 * @param[in]    timeout #3: Timeout value for quit the connection.
 *
 * @retval       -1      #1: select fail
 * @retval       -2      #2: connect fail
 * @retval       -3      #3: connect time out
 * @retval       fd      #4: socket fd that is connected to the remote server.
 *
 * @note         errno is set if connection fail.
 */
int
gf_connect(const char *host, int port, int timeout)
{
    char IP[64];
    struct in_addr addr;
    int ret = inet_pton(AF_INET, host, (void*)&addr);
    if (ret > 0) {
        strcpy(IP, host);
    } else {
        struct hostent *hostEnt;
        hostEnt = gf_gethostbyname(host);
        if (NULL == hostEnt) {
            return (-2);
        }
        strcpy(IP, inet_ntoa(*(struct in_addr*)hostEnt->h_addr_list[0]));
    }

    struct sockaddr_in servaddr;
    servaddr.sin_family = AF_INET;
    servaddr.sin_addr.s_addr = inet_addr(IP);
    servaddr.sin_port = htons(port);

    int clientfd;
    clientfd = socket(AF_INET, SOCK_STREAM, 0);

 retry:
    gf_set_nonblocking(clientfd);
    if(connect(clientfd, (struct sockaddr*)&servaddr, sizeof(struct sockaddr_in)) < 0) {
        int err;
        err = errno;
        if (err == EINTR) {
            goto retry;
	}
        if (err != EINPROGRESS) {
            close(clientfd);
            return (-2);
        }

        time_t start = time(NULL);
        while(1) {
            time_t now = time(NULL);
            if (now - start > timeout) {
                close(clientfd);
                return (-3);
            }

            struct timeval tv; 
            fd_set writefds; 
            tv.tv_sec = timeout; 
            tv.tv_usec = 0; 
            FD_ZERO(&writefds); 
            FD_SET(clientfd, &writefds);
            if (select(clientfd+1, NULL, &writefds, NULL, &tv) <= 0) { 
                continue;
            }

            int sockError=0;
            socklen_t len=sizeof(socklen_t);
            getsockopt(clientfd, SOL_SOCKET, SO_ERROR, (char *)&sockError, &len); 
            if (sockError != 0) {
                close(clientfd);
                errno = sockError;
                return (-2);
            }

            break;
        }
    }

    struct linger   linstr = {1, 10};
    setsockopt(clientfd, SOL_SOCKET, SO_LINGER, (char *)&linstr, sizeof(linstr));

    return (clientfd);
} /* gf_connect */

/**
 * @brief        Connet to remote servers.
 *
 * @param[in]    hosts   #1: A list of the host names to be connected.
 * @param[in]    port    #2: The destinition port.
 * @param[in]    timeout #3: Timeout value for quit the connection.
 *
 * @retval       -1      #1: Failed
 * @retval       fd      #2: socket fd that is connected to the remote server.
 *
 * @note         errno is set if failed to connect to all the hosts.
 */
int
gf_connect_list(const char *hosts, int port, int timeout)
{
    char *hostlist = copy_string(hosts);
    char *list = hostlist;
    char *host;
    char *savedHost;
    int fd = -1;
    while ((host = strtok_r(hostlist, " ", &savedHost)) != NULL) {
        hostlist = NULL;

        fd = gf_connect(host, port, timeout);
        if (fd > 0) {
            break;
        }
    }
    FREEUP(list);
    if (fd < 0) {
        return (-1);
    }

    return (fd);
} /* gf_connect_list */

/**
 * @brief        Connet to remote server.
 *
 * @param[in]    name    #1: Address of the host to be connected.
 * @param[in]    namelen #2: The length of the address.
 * @param[in]    timeout #3: Timeout value for quit the connection.
 *
 * @retval       -1      #1: Failed
 * @retval       fd      #2: socket fd that is connected to the remote server.
 *
 * @note         errno is set if connection fail.
 */
int
gf_connect_addr(struct sockaddr *name, int namelen, int timeout)
{
    int clientfd;
    clientfd = socket(AF_INET, SOCK_STREAM, 0);

    struct sockaddr_in clientaddr;
    memset(&clientaddr, 0, sizeof(struct sockaddr_in));
    clientaddr.sin_family  = AF_INET;
    clientaddr.sin_addr.s_addr = htonl(INADDR_ANY);

    int cc;
    cc = bind(clientfd, (struct sockaddr *)&clientaddr, sizeof(struct sockaddr_in));
    if (cc < 0) {
        close(clientfd);
        return -1;
    }

    fcntl(clientfd, F_SETFD, (fcntl(clientfd, F_GETFD) | FD_CLOEXEC));

    gf_set_nonblocking(clientfd);
    if(connect(clientfd, name, namelen) < 0) {
        int err;
        err = errno;
        if (err != EINPROGRESS) {
            close(clientfd);
            return (-1);
        }

        time_t start = time(NULL);
        while(1) {
            time_t now = time(NULL);
            if (now - start > timeout) {
                close(clientfd);
                return (-1);
            }

            struct timeval tv; 
            fd_set writefds; 
            tv.tv_sec = timeout; 
            tv.tv_usec = 0; 
            FD_ZERO(&writefds); 
            FD_SET(clientfd, &writefds);
            if (select(clientfd+1, NULL, &writefds, NULL, &tv) > 0) { 
                int sockError=0;
                socklen_t len=sizeof(socklen_t);
                getsockopt(clientfd, SOL_SOCKET, SO_ERROR, (char *)&sockError, &len); 
                if (sockError != 0) {
                    close(clientfd);
                    return (-1);
                }
                break;
            }
        }
    }

    struct linger   linstr = {1, 10};
    setsockopt(clientfd, SOL_SOCKET, SO_LINGER,
               (char *)&linstr, sizeof(linstr));

    return (clientfd);
} /* gf_connect_addr */

/**
 * @brief        Receive data from the socket fd and put it to the channel.
 *
 * @param[in]    fd      #1: The fd where to receive data from.
 *
 * @retval       -1      #1: Failed to receive data.
 * @retval        0      #2: Success
 *
 * @note         errno is set if failed to receive data from the fd.
 */
int
gf_recv_data(int fd)
{
    int index = find_chanidx(fd);
    if (index < 0) {
        log_message(ERR, NOCLASS, "%s: cannot find the channel[%d] index.",
                    __func__, fd);
        return (-1);
    }

    struct channel *recvChan = get_channel_at(index);
    if (NULL == recvChan) {
        log_message(ERR, NOCLASS, "%s: cannot find the channel data from index[%d].",
                    __func__, index);
        return (-1);
    }

    struct chan_data *recvData;
    recvData = (struct chan_data *)calloc(1, sizeof(struct chan_data));
    if (NULL == recvData) {
        log_message(ERR, NOCLASS, "%s: cannot allocate(chan_data) %d memory: %m",
                    __func__, sizeof(struct chan_data));
        return (-1);
    }

    char *header = (char *)calloc(HEADER_LEN, sizeof(char));
    if (NULL == header) {
        FREEUP(recvData);
        log_message(ERR, NOCLASS, "%s: cannot allocate(header) %d memory: %m", __func__, HEADER_LEN);
        return (-1);
    }

    unsigned int recvLen = gf_chan_read(recvChan->fd, header, HEADER_LEN, 0);
    if (recvLen != HEADER_LEN) {
        FREEUP(recvData);
        FREEUP(header);
        if (recvLen == 0) {
            return (0);
        }

        log_message(ERR, NOCLASS, "%s: read xdr header failed on channel[%d].", __func__, recvChan->fd);
        return (-1);
    }

    XDR xdrs;
    xdrmem_create(&xdrs, header, sizeof(struct header), XDR_DECODE);
    if (!gf_xdr_header(&xdrs, &recvData->hdr)) {
        FREEUP(recvData);
        FREEUP(header);
        log_message(ERR, NOCLASS, "%s: decode header failed on channel[%d].", __func__, recvChan->fd);
        return (-1);
    }
    FREEUP(header);

    if (recvData->hdr.length > 0) {
        recvData->data = (char *)calloc(recvData->hdr.length, sizeof(char));
        if (NULL == recvData->data) {
            log_message(ERR, NOCLASS, "%s: cannot allocate(data) %d memory: %m",
                        __func__, recvData->hdr.length);

            FREEUP(recvData);
            FREEUP(header);
            return (-1);
        }

        unsigned int dataLen;
        dataLen = gf_chan_read(recvChan->fd, recvData->data, recvData->hdr.length, 0);
        if (dataLen != recvData->hdr.length) {
            FREEUP(recvData);
            FREEUP(header);
            log_message(ERR, NOCLASS, "%s: read channel[%d] data failed.", __func__, fd);
            return (-1);
	}
        recvData->datalen = dataLen;
    }

    pthread_mutex_lock(&mutexChannels);
    if (recvChan->state != CHANNEL_LONGCONN) {
        recvChan->state = CHANNEL_RECV;
    }
    gf_list_insert(recvChan->recvdatalist, recvChan->recvdatalist->forw, (LIST_T *)recvData);
    pthread_mutex_unlock(&mutexChannels);

    return (recvLen + recvData->datalen);
} /* gf_recv_data */

/**
 * @brief        Send the channel data out.
 *
 * @param[in]    fd      #1: The fd where to send the data.
 *
 * @retval       -1      #1: Failed to send data.
 * @retval        0      #2: Success
 *
 * @note         errno is set if failed to send the data out.
 */
int
gf_send_data(int fd)
{
    int index = find_chanidx(fd);
    if (index < 0) {
        return (-1);
    }

    struct channel *sendChan = get_channel_at(index);
    if (sendChan == NULL) {
        return (-1);
    }

    if (LIST_NUM_ENTS(sendChan->senddatalist) == 0) {
        return (-1);
    }

    struct chan_data *sendData;
    for (sendData = (struct chan_data *)sendChan->senddatalist->back;
         sendData != (struct chan_data *)sendChan->senddatalist;
         sendData = sendData->back) {
        unsigned int sentLen = gf_chan_write(sendChan->fd, sendData->data,
                                             sendData->datalen, 5);
        if (sentLen != sendData->datalen) {
            return (-1);
        }
    }

    pthread_mutex_lock(&mutexChannels);
    if (sendChan->state != CHANNEL_LONGCONN) {
        sendChan->state = CHANNEL_SEND;
    }
    pthread_mutex_unlock(&mutexChannels);

    return (0);
} /* gf_send_data */

int
gf_chan_state(int fd, enum chan_state state) 
{
    int index = find_chanidx(fd);
    if (index < 0) {
        return (-1);
    }

    struct channel *chan = get_channel_at(index);
    if (chan == NULL) {
        return (-1);
    }

    struct linger linger;
    linger.l_onoff = 1;
    linger.l_linger = 60;

    pthread_mutex_lock(&mutexChannels);
    chan->state = state;
    if (CHANNEL_LONGCONN == chan->state) {
        if (setsockopt(chan->fd, SOL_SOCKET, SO_LINGER, (char *)&linger,
                       sizeof(struct linger)) < 0) {
            log_message(ERR, NOCLASS, "%s: setsockopt on socket=<%d> option SO_LINGER failed: %m",
                        __func__, chan->fd);
        }
    }
    pthread_mutex_unlock(&mutexChannels);

    return (0);
}

int
gf_get_chan_state(int fd) 
{
    int index = find_chanidx(fd);
    if (index < 0) {
        return (-1);
    }

    struct channel *chan = get_channel_at(index);
    if (chan == NULL) {
        return (-1);
    }

    return chan->state;
}

/**
 * @brief        Go though the channel list to get the channels which have received data.
 *
 * @param[out]   fdlist      #1: The list of fds that have received data.
 *
 * @retval       0           #1: No channel has received data.
 * @retval       number      #2: The number of channels that have received data.
 *
 * @note         the list of fd should be freed by the caller.
 */
unsigned int
gf_get_fdlist(int **fdlist)
{
    unsigned numFd = 0;

    if (NULL == fdlist) {
        return (0);
    }

    *fdlist = (int *)calloc(maxchannels, sizeof(int));
    if (NULL == *fdlist) {
        log_message(ERR, NOCLASS, "%s: cannot allocate(data) %d memory: %m",
                    __func__, maxchannels*sizeof(int));
        return (0);
    }

    pthread_mutex_lock(&mutexChannels);
    int i;
    for (i = 0; i < maxchannels; i++) {
        if (channels[i].state != CHANNEL_RECV && channels[i].state != CHANNEL_LONGCONN) {
            continue;
       }
        (*fdlist)[numFd++] = channels[i].fd;
    }
    pthread_mutex_unlock(&mutexChannels);

    return (numFd);
} /* gf_get_fdlist */

/**
 * @brief        Add data to be sent to the channel.
 *
 * @param[in]    fd          #1: The fd of the channel.
 * @param[in]    data        #2: The data to be sent.
 * @param[in]    length      #3: The length of the data.
 *
 * @retval       -1          #1: Fail
 * @retval        0          #2: Success
 *
 * @note         none.
 */
int
gf_add_data(int fd, char *data, unsigned int length)
{
    int index = find_chanidx(fd);
    if (index < 0) {
        return (-1);
    }

    struct channel *chan = get_channel_at(index);
    if (chan == NULL) {
        return (-1);
    }

    pthread_mutex_lock(&mutexChannels);
    struct chan_data *sendData = (struct chan_data *)calloc(1, sizeof(struct chan_data));
    if (NULL == sendData) {
        log_message(ERR, NOCLASS, "%s: cannot allocate(data) %d memory: %m",
                    __func__, sizeof(struct chan_data));
        return (-1);
    }

    sendData->data = data;
    sendData->datalen = length;
    sendData->stream = OUT;
    gf_list_insert(chan->senddatalist, chan->senddatalist->forw, (LIST_T *)sendData);
    pthread_mutex_unlock(&mutexChannels);

    return (0);
} /* gf_add_data */

/**
 * @brief        Get the data from the channel.
 *
 * @param[in]    fd          #1: The fd of the channel.
 * @param[out]   data        #2: The received data.
 * @param[out]   outhdr      #3: The header of the data.
 *
 * @retval       -1          #1: Fail
 * @retval        0          #2: Success
 *
 * @note         none.
 */
int
gf_get_data(int fd, LIST_T **data)
{
    if (NULL == data || fd < 0) {
        return (-1);
    }

    int index = find_chanidx(fd);
    if (index < 0) {
        return (-1);
    }

    struct channel *chan = get_channel_at(index);
    if (NULL == chan) {
        return (-1);
    }

    pthread_mutex_lock(&mutexChannels);
    int num=LIST_NUM_ENTS(chan->recvdatalist);
    *data = chan->recvdatalist;
    pthread_mutex_unlock(&mutexChannels);

    return (num);
} /* gf_get_data */

/**
 * @brief        check if the given fd is long conn or not.
 *
 * @param[in]    fd          #1: The fd of the channel.
 *
 * @retval       FALSE          #1: Not
 * @retval       TRUE           #2: Is
 *
 * @note         none.
 */
bool_t
gf_check_longconn(int fd)
{
    int index = find_chanidx(fd);
    if (index < 0) {
        return FALSE;
    }

    struct channel *chan = get_channel_at(index);
    if (chan == NULL) {
        return FALSE;
    }

    if (chan->state == CHANNEL_LONGCONN) {
        return TRUE;
    }

    return FALSE;
}

/**
 * @brief        Clean the recvdata from the channel.
 *
 * @param[in]    fd          #1: The fd of the channel.
 *
 * @retval       -1          #1: Fail
 * @retval        0          #2: Success
 *
 * @note         none.
 */
int
gf_clean_recvdata(int fd, struct chan_data *chanData)
{
    int index = find_chanidx(fd);
    if (index < 0) {
        return -1;
    }

    struct channel *chan = get_channel_at(index);
    if (chan == NULL) {
        return (-1);
    }

    pthread_mutex_lock(&mutexChannels);

    gf_list_remove(chan->recvdatalist, (LIST_T *)chanData);
    FREEUP(chanData->data);
    FREEUP(chanData);
    pthread_mutex_unlock(&mutexChannels);

    return 0;
}

/**
 * @brief        To handle socket
 *
 * @param[in]    timeout      #1: timeout of epoll events wait.
 *                                 0: no wait
 *                                -1: wait until events happen
 *
 * @retval       -1      #1: Failed
 * @retval        0      #2: Success
 *
 * @note         Wait for poll events and handle them one by one.
 */
int
gf_handle_socket(int timeout)
{
    int i, connfd, sockfd, retry = 0;
    struct sockaddr_in clientaddr;
    socklen_t clilen = sizeof(struct sockaddr_in);

    int nfds = 0;
    while (retry < 3) {
        nfds = epoll_wait(epfd, events, gf_max_channels(), timeout);
        if (nfds < 0) {
            if (errno == EINTR) {
                retry++;
                continue;
            }

            log_message(ERR, NOCLASS, "%s: wait for poll failed: %m.", __func__);
            return -1;
        }

        break;
    }

    for (i=0; i<nfds; i++) {
        if(events[i].data.fd == listenfd){
            connfd = accept(listenfd, (struct sockaddr *)&clientaddr, &clilen);
            if(connfd<0){
                continue;
            }
            gf_set_nonblocking(connfd);
            //char *ip = inet_ntoa(clientaddr.sin_addr);
            //int port = ntohs(clientaddr.sin_port);

            if (gf_new_conn(connfd) < 0) {
                /* cannot find free fd, wait a while */
                log_message(ERR, NOCLASS, "%s: no available fd now, wait for a while.", __func__);
                close(connfd);
            }

            struct epoll_event ev;

            memset(&ev, 0, sizeof(struct epoll_event));
            ev.data.fd=connfd;
            ev.events=EPOLLIN|EPOLLET;
            epoll_ctl(epfd, EPOLL_CTL_ADD,connfd,&ev);

            /* reset epoll event for listenfd */
            memset(&ev, 0, sizeof(struct epoll_event));
            ev.data.fd = listenfd;
            ev.events = EPOLLIN|EPOLLET;
            epoll_ctl(epfd, EPOLL_CTL_MOD, listenfd, &ev);            
        } else {
            if ((sockfd = events[i].data.fd) < 0) {
                continue;
            }

            if(events[i].events & (EPOLLRDHUP | EPOLLERR)) {
                epoll_ctl(epfd, EPOLL_CTL_DEL, sockfd, NULL);
                gf_chan_close(sockfd);
                continue;
            }

            int ret;
            if (events[i].events & EPOLLIN) {
                ret = gf_recv_data(sockfd);
                if (ret < 0) {
                    log_message(ERR, NOCLASS, "%s: gf_recv_data failed on channel[%d].",
                                __func__, sockfd);

                    epoll_ctl(epfd, EPOLL_CTL_DEL, sockfd, NULL);
                    gf_chan_close(sockfd);
                    continue;
                } else if (ret == 0) {
                    log_message(ERR, NOCLASS, "%s: gf_recv_data timeout on channel[%d].",
                                __func__, sockfd);

                    epoll_ctl(epfd, EPOLL_CTL_DEL, sockfd, NULL);
                    gf_chan_close(sockfd);

                    continue;
                }
            } else if(events[i].events & EPOLLOUT) {
                ret = gf_send_data(sockfd);
                if (ret < 0) {
                    log_message(ERR, NOCLASS, "%s: gf_send_data failed on channel[%d].",
                                __func__, sockfd);
                }

                struct channel *chan = NULL;
                int index = find_chanidx(sockfd);
                if (index >= 0) {
                    chan = get_channel_at(index);
                    if (chan != NULL && chan->state == CHANNEL_LONGCONN) {

                        gf_list_free(chan->senddatalist, free_data);
                        chan->senddatalist = gf_list_make("Channal sending data list");

                        gf_list_free(chan->recvdatalist, free_data);
                        chan->recvdatalist = gf_list_make("Channal receiving data list");

                        struct epoll_event ev;
                        memset(&ev, 0, sizeof(struct epoll_event));
                        ev.data.fd = sockfd;
                        ev.events = EPOLLIN;
                        epoll_ctl(epfd, EPOLL_CTL_MOD, sockfd, &ev);

                        continue;
                    }
                }

                struct epoll_event ev;
                memset(&ev, 0, sizeof(struct epoll_event));
                ev.data.fd = sockfd;
                ev.events = EPOLLRDHUP;
                epoll_ctl(epfd, EPOLL_CTL_MOD, sockfd, &ev);
            }
        }
    }
    
    return 0;
} /* gf_handle_socket */

/**
 * @brief        To send back the header to requester right away
 *
 * @retval       -1      #1: Failed
 * @retval        0      #2: Success
 *
 * @note         None.
 */
int
gf_sendback_header2(int fd, struct header *outhdr)
{
    unsigned int replySize = 0;
    replySize += HEADER_LEN;
    char *replyBuf = (char *)calloc(replySize, sizeof(char));
    if (NULL == replyBuf) {
        log_message(ERR, NOCLASS, "%s: cannot allocate(data) %d memory: %m",
                    __func__, replySize*sizeof(char));
        return (-1);
    }

    XDR xdrsend;
    xdrmem_create(&xdrsend, replyBuf, replySize, XDR_ENCODE);
    if (!gf_xdr_message(&xdrsend, NULL, outhdr, NULL, NULL, NULL)) {
        xdr_destroy(&xdrsend);
        FREEUP(replyBuf);
        return (-1);
    }
    xdr_destroy(&xdrsend);

    replySize = outhdr->length + HEADER_LEN;
    gf_add_data(fd, replyBuf, replySize);
    gf_send_data(fd);

    epoll_ctl(epfd, EPOLL_CTL_DEL, fd, NULL);

    gf_chan_close(fd);

    return 0;
}

/**
 * @brief        To send back the header to requester.
 *
 * @retval       -1      #1: Failed
 * @retval        0      #2: Success
 *
 * @note         None.
 */
int
gf_sendback_header(int fd, struct header *outhdr)
{
    unsigned int replySize = 0;
    replySize += HEADER_LEN;
    char *replyBuf = (char *)calloc(replySize, sizeof(char));
    if (NULL == replyBuf) {
        log_message(ERR, NOCLASS, "%s: cannot allocate(data) %d memory: %m",
                    __func__, replySize*sizeof(char));
        return (-1);
    }

    XDR xdrsend;
    xdrmem_create(&xdrsend, replyBuf, replySize, XDR_ENCODE);
    if (!gf_xdr_message(&xdrsend, NULL, outhdr, NULL, NULL, NULL)) {
        xdr_destroy(&xdrsend);
        FREEUP(replyBuf);
        return (-1);
    }
    xdr_destroy(&xdrsend);

    replySize = outhdr->length + HEADER_LEN;
    gf_add_data(fd, replyBuf, replySize);

    struct epoll_event ev;
    memset(&ev, 0, sizeof(struct epoll_event));
    ev.data.fd = fd;
    ev.events = EPOLLOUT | EPOLLET;
    epoll_ctl(epfd, EPOLL_CTL_MOD, fd, &ev);

    return (0);
} /* gf_sendback_header */

/**
 * @brief        To send back the header to requester.
 *
 * @retval       -1      #1: Failed
 * @retval        0      #2: Success
 *
 * @note         None.
 */
int
gf_errorback(int fd, struct header *outhdr, char *message)
{
    unsigned int replySize = 0;
    replySize += HEADER_LEN + strlen(message) + 100;
    char *replyBuf = (char *)calloc(replySize, sizeof(char));
    if (NULL == replyBuf) {
        log_message(ERR, NOCLASS, "%s: cannot allocate(data) %d memory: %m",
                    __func__, replySize*sizeof(char));
        return (-1);
    }

    XDR xdrsend;
    xdrmem_create(&xdrsend, replyBuf, replySize, XDR_ENCODE);
    if (!gf_xdr_message(&xdrsend, &message, outhdr, NULL, gf_xdr_string, NULL)) {
        xdr_destroy(&xdrsend);
        FREEUP(replyBuf);
        return (-1);
    }
    xdr_destroy(&xdrsend);

    replySize = outhdr->length + HEADER_LEN;
    gf_add_data(fd, replyBuf, replySize);

    struct epoll_event ev;
    memset(&ev, 0, sizeof(struct epoll_event));
    ev.data.fd = fd;
    ev.events = EPOLLOUT | EPOLLET;
    epoll_ctl(epfd, EPOLL_CTL_MOD, fd, &ev);

    return (0);
} /* gf_errorback */

/**
 * @brief        Call master to send request and get the reply.
 *
 * @param[in]    masterlist  #1: The list of master hosts.
 * @param[in]    port        #2: port to be connected.
 * @param[in]    connTimeout #3: connection time out.
 * @param[in]    recvTimeout #4: receive data time out.
 * @param[in]    requestSize #5: the length of request data.
 * @param[in]    request     #6: the request data.
 * @param[out]   ourhdr      #7: the header of the reply.
 * @param[out]   reply       #8: the data of the reply.
 * @param[out]   fd          #8: the fd to communicate with master.
 *
 * @retval       -1          #1: Fail
 * @retval        0          #2: Success
 *
 * @note         none.
 */
int
gf_callmaster(const char *masterlist, int port, int connTimeout, int recvTimeout, int retry,
              unsigned int requestlen, const char *request,
              struct header *outhdr, char **reply, int *fd)
{
    if (fd == NULL) {
        log_message(ERR, NOCLASS, "%s: input parameter is null(fd)");
        return (-1);
    }

    if (reply) {
        *reply = NULL;
    }

    *fd = -1;
    
    char *hostlist = copy_string(masterlist);
    char *list = hostlist;
    char *host;
    char *savedHost;
    int cc = 0;
    while ((host = strtok_r(hostlist, " ", &savedHost)) != NULL) {
        hostlist = NULL;

        int i;
        for (i=0; i<retry; i++) {

            *fd = gf_connect(host, port, connTimeout);
            if (*fd < 0) {
                gf_sleep(500);
                continue;
            } else {
                break;
            }
        }

        if (i == retry && *fd < 0) {
            continue;
	}

        if (gf_new_conn(*fd) < 0) {
            /* cannot find free fd, wait a while */
            log_message(ERR, NOCLASS, "%s: no available fd now, wait for a while.", __func__);
            close(*fd);
            *fd = -1;
            FREEUP(list);
            return (-1);
        }

        unsigned int sendLen;
        sendLen = gf_chan_write(*fd, request, requestlen, connTimeout);
        if (sendLen != requestlen) {
            gf_chan_close(*fd);
            *fd = -1;
            FREEUP(list);
            return (-1);
        }

        pthread_mutex_lock(&mutexMasterName);
        if (masterName == NULL || strcmp(masterName, host) != 0) {
            FREEUP(masterName);
            masterName = copy_string(host);
        }
        pthread_mutex_unlock(&mutexMasterName);

        /* if reply == NULL, it implies that no need to get reply from the peer.
         * we will consider later if it requires one more paramter, such as option,
         * to indicate no reply mode.
         */
        if (reply == NULL || outhdr == NULL) {
            struct epoll_event ev;

            memset(&ev, 0, sizeof(struct epoll_event));
            ev.data.fd=*fd;
            ev.events=EPOLLIN|EPOLLET;
            epoll_ctl(epfd, EPOLL_CTL_ADD,*fd,&ev);            
            break;
        }

        gf_init_header(outhdr);
        cc = gf_read_reply(*fd, recvTimeout, outhdr, reply);
        if (cc <= 0) {
            if (outhdr->reqcode == 0) {
                gf_chan_close(*fd);
                continue;
            }
        }

        break;
    }
    FREEUP(list);

    if (*fd < 0 || cc < 0) {

        if (*fd < 0) {
            if (-1 == *fd) {
                log_message(ERR, NOCLASS, "%s: cannot get all master node name.", __func__);
            } else if (-2 == *fd) {
                log_message(ERR, NOCLASS, "%s: connect to master hosts[%s] failed: %m.",
                            __func__, masterlist);
            } else if (-3 == *fd) {
                log_message(ERR, NOCLASS, "%s: connect to master hosts[%s] timeout.",
                            __func__, masterlist);
            }
            return *fd;
        }
        return (-1);
    }

    return (0);
} /* gf_callmaster */


char *
gf_get_master(void)
{
    return masterName;
}

/**
 * @brief        Call slave to send request and get the reply.
 *
 * @param[in]    slave       #1: The slave host name.
 * @param[in]    port        #2: port to be connected.
 * @param[in]    connTimeout #3: connection time out.
 * @param[in]    recvTimeout #4: receive data time out.
 * @param[in]    requestSize #5: the length of request data.
 * @param[in]    request     #6: the request data.
 * @param[out]   ourhdr      #7: the header of the reply.
 * @param[out]   reply       #8: the data of the reply.
 *
 * @retval       -1          #1: Fail
 * @retval       -2          #2: connect fail
 * @retval       -3          #3: connect time out
 * @retval       -4          #4: send request fail
 * @retval       -5          #5: receive reply fail
 * @retval        0          #6: Success
 *
 * @note         none.
 */
int
gf_callslave(const char *slave, int port, int connTimeout, int recvTimeout,
             unsigned int requestSize, const char *request,
             struct header *outhdr, char **reply, int *fd)
{
    if (fd == NULL) {
        return (-1);
    }

    if (reply) {
        *reply = NULL;
    }

    int cc = 0;
    if (*fd < 0) {
        *fd = gf_connect(slave, port, connTimeout);
        if (*fd < 0) {
            if (-1 == *fd) {
                log_message(ERR, NOCLASS, "%s: cannot get my host name [%s].", __func__, slave);
            } else if (-2 == *fd) {
                if (0 == errno) {
                    log_message(ERR, NOCLASS, "%s: connect to host[%s] failed.", __func__, slave);
                }else {
                    log_message(ERR, NOCLASS, "%s: connect to host[%s] failed: %m.", __func__, slave);
                }
            } else if (-3 == *fd) {
                log_message(ERR, NOCLASS, "%s: connect to host[%s] timeout.", __func__, slave);
            }
            return (*fd);
        }

        if (gf_new_conn(*fd) < 0) {
            /* cannot find free fd, wait a while */
            log_message(ERR, NOCLASS, "%s: no available fd now, wait for a while.", __func__);
            close(*fd);
            *fd = -1;
            return (-6);
        }
    }

    unsigned int sendLen;
    sendLen = gf_chan_write(*fd, request, requestSize, connTimeout);
    if (sendLen != requestSize) {
        gf_chan_close(*fd);
        *fd = -1;
        return (-4);
    }

    /* if reply == NULL, it implies that no need to get reply from the peer.
     * we will consider later if it requires one more paramter, such as option,
     * to indicate no reply mode.
     */
    if (reply == NULL || outhdr == NULL) {
        struct epoll_event ev;

        memset(&ev, 0, sizeof(struct epoll_event));
        ev.data.fd=*fd;
        ev.events=EPOLLIN|EPOLLET;
        epoll_ctl(epfd, EPOLL_CTL_ADD,*fd,&ev);
            
        return (0);
    }

    gf_init_header(outhdr);
    cc = gf_read_reply(*fd, recvTimeout, outhdr, reply);
    if (cc < 0) {
        gf_chan_close(*fd);
        *fd = -1;
        return (-5);
    }

    return (0);
} /* gf_callslave */

/**
 * @brief        To receive the reply data from the remote server.
 *
 * @param[in]    fd          #1: The fd to receive more data from.
 * @param[in]    recvTimeout #4: receive data time out.
 * @param[out]   ourhdr      #7: the header of the reply.
 * @param[out]   reply       #8: the data of the reply.
 *
 * @retval       -1          #1: Fail
 * @retval        0          #2: Success
 *
 * @note         none.
 */
int
gf_read_reply(int fd,  int recvTimeout, struct header *outhdr, char **reply)
{
    char *header = (char *)calloc(HEADER_LEN, sizeof(char));
    if (NULL == header) {
        gf_chan_close(fd);
        log_message(ERR, NOCLASS, "%s: cannot allocate(data) %d memory: %m",
                    __func__, HEADER_LEN*sizeof(char));
        return (-1);
    }

    unsigned int recvLen = gf_chan_read(fd, header, HEADER_LEN, recvTimeout);
    if (recvLen != HEADER_LEN) {
        FREEUP(header);
        gf_chan_close(fd);
        return (-1);
    }

    XDR xdrs;
    xdrmem_create(&xdrs, header,  sizeof(struct header), XDR_DECODE);
    if (!gf_xdr_header(&xdrs, outhdr)) {
       FREEUP(header);
       gf_chan_close(fd);
       return (-1);
    }
    FREEUP(header);
    xdr_destroy(&xdrs);

    if (outhdr->length > 0) {

        *reply = (char *)calloc(outhdr->length, sizeof(char));
        if (NULL == *reply) {
            gf_chan_close(fd);
            log_message(ERR, NOCLASS, "%s: cannot allocate(data) %d memory: %m",
                        __func__, outhdr->length*sizeof(char));
            return (-1);
        }

        unsigned int dataLen;
        dataLen = gf_chan_read(fd, *reply, outhdr->length, recvTimeout);
        if (dataLen != outhdr->length) {
            FREEUP(*reply);
            gf_chan_close(fd);
            return (-1);
	}
    } else {
        *reply = NULL;
    }

    return (outhdr->length);
} /* gf_read_reply */

/**
 * @brief        To set the socket to none blocking.
 *
 * @param[in]    sock          #1: The fd to be set.
 *
 * @note         none.
 */
void 
gf_set_nonblocking(int sock)
{
    int opts;

    opts=fcntl(sock,F_GETFL);
    if(opts<0) {
        return;
    }

    opts |= O_NONBLOCK;
    if(fcntl(sock,F_SETFL,opts)<0) {
        return;
    }

    return;
}

/**
 * @brief        To set the socket to blocking.
 *
 * @param[in]    sock          #1: The fd to be set.
 *
 * @note         none.
 */
void
gf_set_blocking(int sock)
{
    int opts;
    opts=fcntl(sock,F_GETFL);
    if(opts<0) {
        return;
    }
    opts &= ~O_NONBLOCK;
    if(fcntl(sock,F_SETFL,opts)<0) {
        return;
    }

    return;
}

/**
 * @brief        bind address to the socket.
 *
 * @param[in]    sock          #1: The fd to be bound.
 * @param[in]    myaddr        #2: IP address.
 * @param[in]    myaddr        #3: IP address length.
 *
 * @retval       -1          #1: Fail
 * @retval        0          #2: Success
 *
 * @note         none.
 */
int
gf_sock_bind(int sock, struct sockaddr *myaddr, int addrlen)
{
    struct sockaddr_in *cliaddr;
    ushort port;
    int i;

    cliaddr = (struct sockaddr_in *)myaddr;
    if (0 != cliaddr->sin_port) {
        return (bind(sock, myaddr, addrlen));
    }

    for (i = 1; i <= BIND_RETRY_TIMES; i++) {

        if (1 == i) {
            port = (ushort)(time(0) | getpid());
            port = ((port < 1024) ? (port + 1024) : port);
        } else {
            port++;
            port = ((port < 1024) ? (port + 1024) : port);
        }

        cliaddr->sin_port = htons(port);

        if (bind(sock, (struct sockaddr *)cliaddr, addrlen) == 0) {
            return 0;
        } else if (errno == EADDRINUSE) {
            continue;
        } else {
            return -1;
        }
    }

    return -1;
} /* gf_sock_bind */

/**
 * @brief        Transfer sockaddr_in to string.
 *
 * @param[in]    from          #1: The sock address.
 * @param[out]   adbuf         #2: The string of the sock address.
 *
 * @retval       The string of the sock address    #1
 */
char *
gf_sockaddr_str(struct sockaddr_in *from, char *adbuf)
{
    snprintf(adbuf, 24, "%s:%hu", inet_ntoa(from->sin_addr), ntohs(from->sin_port));

    return (adbuf);
}

void
gf_sleep(int msec)
{
    struct timeval tv;

    if (msec < 1) {
	return;
    }

    tv.tv_sec = msec /1000;
    tv.tv_usec = (msec - tv.tv_sec * 1000) * 1000;

    select(0, 0, 0, 0, &tv);

    return;
}

/**
 * @brief        To get the maximum number of channels in the system.
 *
 * @retval       number of maximum channel          #1
 *
 * @note         none.
 */
int
gf_max_channels(void)
{
    if (maxchannels <= 0) {
        maxchannels = 1024;
    }

    return maxchannels;
};

static void 
free_data(void *data)
{
    struct chan_data *chanData = (struct chan_data *)data;
    if (chanData == NULL) {
        return;
    }
    FREEUP(chanData->data);
    FREEUP(chanData);
    return;
} /* free_data */

static struct channel *
get_free_channel(int *index)
{
    if (index == NULL) {
        return (NULL);
    }

    *index = -1;

    pthread_mutex_lock(&mutexChannels);
    int i;
    for (i = 0; i < maxchannels; i++) {
        if (channels[i].state == CHANNEL_FREE) {
            channels[i].state = CHANNEL_DISC;
            *index = i;
            break;
        }
    }
    pthread_mutex_unlock(&mutexChannels);

    if (i <= maxchannels) {
        return &(channels[i]);
    } else {
        return (NULL);
    }
}

static struct channel *
get_channel_at(int index)
{
    if (index < 0) {
        return (NULL);
    }

    return (&(channels[index]));
}

unsigned int
gf_chan_read(int s, char *buf, unsigned int len, int timeout)
{
    int cc;
    int nReady;
    unsigned int length = len;
    unsigned int ret = 0;

    time_t start = time(NULL);

    while(1) {
        time_t now = time(NULL);
        if (now - start > timeout) {
            log_message(ERR, NOCLASS, "%s: read data timeout on channel[%d].", __func__, s);
            goto End;
        }

        struct timeval timeval;
        timeval.tv_sec  = 1;
        timeval.tv_usec = 0;

        nReady = gf_read_select(s, timeval);
        if (nReady < 0) {
            log_message(ERR, NOCLASS, "%s: select failed on channel[%d] due to: %m.", __func__, s);
            goto End;
        } else if (nReady == 0) {
            continue;
        } else {
            if ((cc = read(s, (char *)buf, len)) > 0) {
                len -= cc;
                buf += cc;
            } else if (cc < 0) {
                if (VALID_IO_ERR(errno)) {
                    continue;
                }
                log_message(ERR, NOCLASS, "%s: read failed on channel[%d] due to: %m.", __func__, s);
                goto End;
            } else {
                if (errno == EINPROGRESS) {
                    continue;
                }
                /* if (cc == 0) peer close connection, regarded as ok */
                break;
            }

            if (len == 0 ) {
                break;
            }
        }
    }

    ret = length - len;

End:
    return ret;
} /* gf_chan_read */

unsigned int 
gf_chan_write(int s, const char *buf, unsigned int len, int timeout)
{
    int cc;
    unsigned int length;
    unsigned int ret = 0;

    time_t start = time(NULL), now;

    for (length = len; len > 0; ) {
        if ((cc = write(s, (char *)buf, len)) > 0) {
            len -= cc;
            buf += cc;
        } else if (cc < 0) {
            if (VALID_IO_ERR(errno)) {
            } else {
                log_message(ERR, NOCLASS, "%s: write %d/%d bytes on channel[%d] failed due to: %m.",
                            __func__, len, length, s);
                goto End;
            }
        }

        if (len > 0) {   /* not done yet */
            now = time(NULL);
            if (now - start > timeout) {
                log_message(ERR, NOCLASS, "%s: timeout.", __func__);
                goto End;
            }
        }
    }
End:
    ret = length - len;
    return ret;
} /* gf_chan_write */

int 
gf_read_select(int s, struct timeval timeout)
{
    int cc;

    if (s >= 1024) {
        struct pollfd   p;
        int             n;
        int             t;

        n = 1;
        p.fd = s;
        p.events = 0;

        p.events |= POLLIN;

        while (1) {
            if (timeout.tv_sec == 0) {
                t = -1;
            } else {
                t = timeout.tv_sec * 1000;
            }
            p.revents = 0;

            cc = poll(&p, n, t);

            if (cc >= 0) {
                return(cc);
            }

            if (errno == EINTR)
                continue;

            return(cc);
        }
    }

    fd_set rmask;
    while(1) {
        FD_ZERO(&rmask);
        FD_SET(s, &rmask);

        cc = select(s+1, &rmask, NULL, NULL, &timeout);

        if (cc >= 0) {
            return cc;
        }

        if (errno == EINTR) {
            continue;
        }

        return (cc);
    }
} /* gf_read_select */

static int
add_chanidx(struct channel *channel, int index)
{
    if (channel == NULL) {
        return (-1);
    }

    char fdKey[12];
    snprintf(fdKey, 12, "%d", channel->fd);
    int *ent = (int *)calloc(1, sizeof(int));
    if (ent == NULL) {
        return (-1);
    }

    *ent = index;
    pthread_mutex_lock(&mutexChanHash);
    gf_hash_install(chanhash, fdKey, (void *)ent);
    pthread_mutex_unlock(&mutexChanHash);

    return (0);
} /* add_chanidx */

static int
find_chanidx(int fd)
{
    char fdKey[12];
    snprintf(fdKey, 12, "%d", fd);

    int *index;
    pthread_mutex_lock(&mutexChanHash);
    index = (int *)gf_hash_find(chanhash, fdKey);
    pthread_mutex_unlock(&mutexChanHash);
    if (index == NULL) {
        return -1;
    }

    return (*index);
} /* find_chanidx */
